package com.netty.handler;

import com.netty.constant.Constants;
import com.netty.helper.ChannelGroupHelper;
import com.netty.helper.MessageFactory;
import com.netty.model.Message;
import com.netty.model.MessageWrapper;
import com.netty.service.ConnectService;
import com.netty.service.MessageService;
import com.netty.util.ImUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

/**
 * @author suzhen
 * @create 2018/7/13
 */
@Slf4j
public class MessageRequestHandler extends SimpleChannelInboundHandler<Message> {

    private ConnectService connectService;
    private MessageService messageService;

    public MessageRequestHandler(MessageService messageService, ConnectService connectService) {
        this.messageService = messageService;
        this.connectService = connectService;
    }

    /**
     * 心跳检测，此方法会被定时执行
     *
     * @param ctx
     * @param evt
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        String sessionId = ctx.channel().attr(Constants.SessionConfig.SERVER_SESSION_ID).get();
        //发送心跳包
        if (evt instanceof IdleStateEvent && ((IdleStateEvent) evt).state().equals(IdleState.WRITER_IDLE)) {
            if (StringUtils.isNotEmpty(sessionId)) {
                log.debug("心跳检测");
                Message message = MessageFactory.createMessage(sessionId);
                message.setCmd(Constants.CmdType.HEARTBEAT);
                message.setType(Constants.ProtobufType.SEND);
                ctx.channel().writeAndFlush(message);
            }
            log.debug(IdleState.WRITER_IDLE + "... from " + sessionId + "-->" + ctx.channel().remoteAddress() + " nid:" + ctx.channel().id().asShortText());
        }

        //如果心跳请求发出70秒内没收到响应，则关闭连接
        if (evt instanceof IdleStateEvent && ((IdleStateEvent) evt).state().equals(IdleState.READER_IDLE)) {
            log.debug(IdleState.READER_IDLE + "... from " + sessionId + " nid:" + ctx.channel().id().asShortText());
            Long lastTime = (Long) ctx.channel().attr(Constants.SessionConfig.SERVER_SESSION_HEARTBEAT).get();
            if (lastTime == null || ((System.currentTimeMillis() - lastTime) >= Constants.ImserverConfig.PING_TIME_OUT)) {
                //关闭，用户退出,用户在规定时间内没有活动就踢出用户
                connectService.close(ctx);
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
        log.warn("ImWebSocketServerHandler (" + ImUtils.getRemoteAddress(ctx) + ") -> Unexpected exception from downstream." + cause);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        ChannelGroupHelper.add(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        ChannelGroupHelper.remove(ctx.channel());
    }

    @Override
    protected void messageReceived(ChannelHandlerContext ctx, Message message) {
        String sessionId = message.getSender();
        log.debug("messageReceived: {}", message);

        // inbound
        if (message.getType() == Constants.ProtobufType.SEND) {
            //更新心跳时间
            ctx.channel().attr(Constants.SessionConfig.SERVER_SESSION_HEARTBEAT).set(System.currentTimeMillis());
        }

        // outbound
        log.debug("message.type: {}", message.getType());
        log.debug("message.CmdType: {}", Constants.CmdType.MESSAGE);
        if (message.getType() == Constants.ProtobufType.SEND || message.getType() == Constants.ProtobufType.REPLY) {
            log.debug("messageReceived : MESSAGE");
            MessageWrapper wrapper = messageService.convertToMessageWrapper(sessionId, message);
            if (wrapper != null) {
                receiveMessages(ctx, wrapper);
            }
        }
    }

    /**
     * to send message
     *
     * @param ctx
     * @param wrapper
     */
    private void receiveMessages(ChannelHandlerContext ctx, MessageWrapper wrapper) {
        //设置消息来源为Websocket
        wrapper.setSource(Constants.ImserverConfig.WEBSOCKET);
        log.debug("receiveMessages MessageWrapper: {}", wrapper);
        if (wrapper.isConnect()) {
            connectService.connect(ctx, wrapper);
        } else if (wrapper.isHeartbeat()) {
            log.debug("receiveMessages isHeartbeat");
            connectService.heartbeatToClient(ctx, wrapper);
        } else if (wrapper.isGroup()) {
            log.debug("receiveMessages isGroup");
            connectService.pushGroupMessage(wrapper);
        } else if (wrapper.isSend()) {
            log.debug("receiveMessages isSend");
            connectService.pushMessage(ctx, wrapper);
        } else if (wrapper.isReply()) {
            log.debug("receiveMessages isReply");
            connectService.pushMessage(ctx, wrapper.getSessionId(), wrapper);
        } else if (wrapper.isClose()) {
            log.debug("receiveMessages isClose");
            connectService.close(ctx, wrapper);
        }
    }
}
